package com.yanfan.iot.ruleEngine;

import com.alibaba.fastjson2.JSON;
import com.yanfan.framework.utils.RedisUtils;
import com.yanfan.iot.domain.Bridge;
import com.yanfan.iot.domain.MqttClient;
import com.yanfan.iot.service.IBridgeService;
import com.yanfan.iot.service.IMqttBridgeClientService;
import com.yanfan.mqttclient.PubMqttClient;
import com.yomahub.liteflow.annotation.LiteflowComponent;
import com.yomahub.liteflow.core.NodeComponent;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.Duration;
import java.util.Objects;

@Slf4j
@LiteflowComponent("mqttBridge")
public class MqttBridge extends NodeComponent {
    @Autowired
    private IBridgeService bridgeService;

    @Autowired
    private IMqttBridgeClientService mqttClientService;

    @Autowired
    private PubMqttClient mqttClient;

    @Override
    public void process() throws Exception {
        // 获取上下文对象
        MsgContext cxt = this.getContextBean(MsgContext.class);
        Integer id = cxt.getData("mqttBridgeID");
        Bridge bridge = bridgeService.selectBridgeById(Long.valueOf(id));
        if (bridge != null && Objects.equals(bridge.getEnable(), "1")) {
            // 接入数据
            if (bridge.getDirection() == 1) {
//                if (Objects.equals(bridge.getRoute(), cxt.getTopic())) {
                    String cacheKey = "mqttBridge:" + bridge.getId() + ":" + bridge.getRoute();
                    Boolean isSend = RedisUtils.getCacheObject(cacheKey);
                    // 重复的消息不重复执行，1秒之内只允许发送一次，防止桥接的mqtt topic 与 自己的服务器主题一致，导致出现循环
                    if (Objects.isNull(isSend) || !isSend) {
                        mqttClient.publish(0, false, cxt.getTopic(), cxt.getPayload());
                        RedisUtils.setCacheObject(cacheKey, true, Duration.ofSeconds(1));
                    }
//                }
            } else if (bridge.getDirection() == 2) { // 推送数据
                MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                MqttAsyncClient client = MqttBridgeClientFactory.instance(mqttClientService.buildMqttClientConfig(config, bridge.getId()));
                MqttMessage message = new MqttMessage();
                message.setPayload(cxt.getPayload().getBytes());
                String topic = cxt.getTopic();
                try {
                    client.publish(topic, message);
                    log.info("=>桥接发布主题成功 topic={},message={}", topic, message);
                } catch (MqttException e) {
                    log.error("msg:" + e.getMessage());
                    log.error("cause:" + e.getCause());
                    log.error("reason:" + e.getReasonCode());
                    log.error("=>桥接发布主题时发生错误 topic={},message={}", cxt.getTopic(), e.getMessage());
                }
            }

        }
    }
}
